package com.alibaba.otter.node.etl.load.loader.mq.rabbit.api;

import com.alibaba.otter.node.etl.load.loader.mq.rabbit.exception.MqPluginException;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.model.RabbitMqNamespace;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.namespace.NamespaceManager;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.resolver.RabbitMqNamespaceResolver;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.LoadingCache;
import com.rabbitmq.client.Connection;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Slf4j
public class RabbitMqConnectionManager implements ConnectionManager<RabbitMqNamespace, Connection> {

    private static LoadingCache<RabbitMqNamespace, Connection> connCache = CacheBuilder.
            newBuilder().build(new RabbitMqConnectionLoader());


    public void closeConnection(String namespaceName) {
        RabbitMqNamespaceResolver resolver = new RabbitMqNamespaceResolver(namespaceName);
        RabbitMqNamespace namespace = (RabbitMqNamespace) NamespaceManager.get(resolver);
        Connection connection = getConnection(namespaceName);

        // 同步移除内存中的连接
        connCache.invalidate(namespace);
        //  异步关闭连接，关闭连接可能会block很长时间。同步执行会导致线程卡住。
        executorService.execute(() -> closeQuietly(connection));
    }

    private static ExecutorService executorService = Executors.newFixedThreadPool(1);

    public Connection getConnection(String namespaceName) {
        RabbitMqNamespaceResolver resolver = new RabbitMqNamespaceResolver(namespaceName);
        RabbitMqNamespace namespace = (RabbitMqNamespace) NamespaceManager.get(resolver);
        return getConnection(namespace);
    }

    @Override
    public synchronized Connection getConnection(RabbitMqNamespace namespace) {
        try {
            return connCache.get(namespace);
        } catch (ExecutionException e) {
            throw new MqPluginException("创建连接失败", e);
        }
    }

    private void closeQuietly(Connection connection) {
        try {
            connection.close();
        } catch (IOException e) {
            log.error("failed to close rabbitmq connection.", e);
        }
    }
}
